package com.bw.flinkstreaming.state2.job4;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;


public class ReducingStateSum extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Long>> {

    //定义 reduceState
    private ReducingState<Long> avgState;

    /**
     * 初始化state
     * */
    @Override
    public void open(Configuration parameters) throws Exception {
        ReducingStateDescriptor descriptor = new ReducingStateDescriptor<Long>("reduceAvg",
                /**
                 * 参数 total_value：累计的值
                 * 参数 current_value：当前的值
                 * */
                new ReduceFunction<Long>() {
                    @Override
                    public Long reduce(Long total_value, Long current_value) throws Exception {
                        return total_value+current_value;
                    }
                },Long.class);
        avgState = getRuntimeContext().getReducingState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
        avgState.add(value.f1);
        out.collect(Tuple2.of(value.f0,avgState.get()));
    }
}
